package homework5

import akka.actor.{Actor, ActorRef, ActorSystem, Props}
import com.typesafe.config.ConfigFactory
import homework5.Common._

import scala.collection.mutable
import scala.concurrent.duration._


/**
 * master
 */
class SparkMaster extends Actor {

  // 存放注册的worker信息，可变的hashmap
  val workers = mutable.HashMap[String, WorkInfo]()

  override def receive: Receive = {
    // 接收到开始消息
    case StartMessage(msg) => {
      println(s"Spark Master start. $msg")
      // 给自己发消息，开始超时检查
      self ! StartTimeoutCheck
    }
    // 注册消息
    case RegisterWorkInfo(id, cpu, ram) => {
      // master收到worker的注册消息，如果未注册过，完成注册并返回
      if (!workers.contains(id)) {
        // 注册逻辑
        val workInfo = new WorkInfo(id, cpu, ram)
        workers += (id -> workInfo)
        sender ! RegisteredWorkerInfo
        println(s"worker[$id] 完成注册")
      }
    }
    // 心跳消息
    case HeartBeat(id) => {
      workers(id).lasterHeartBeatTime = System.currentTimeMillis()
      println(s"worker[$id]已更新心跳")
    }
    // 开始超时时间检查
    case StartTimeoutCheck => {
      import context.dispatcher
      // master启动定时任务，检测worker
      println("master开启定时检测心跳任务...")
      import context.dispatcher
      context.system.scheduler.schedule(0 millis, 10010 millis, self, RemoveTimeoutWorker)
    }


    // 移除超时的worker
    case RemoveTimeoutWorker => {
      // 心跳检测逻辑
      println("心跳检测中...")
      val current = System.currentTimeMillis()
      val workersInfo = workers.values
      workersInfo.filter(current - _.lasterHeartBeatTime > 6000) // 超过6s没有更新心跳则过期，从注册的workers中移除掉
        .foreach(workersInfo => workers.remove(workersInfo.id))
    }
  }
}

object SparkMaster extends App {
  // 配置信息
  val masterHost = "127.0.0.1"
  val masterPort = 10010

  val config = ConfigFactory.parseString(
    s"""
       |akka.actor.provider="akka.remote.RemoteActorRefProvider"
       |akka.remote.netty.tcp.hostname=$masterHost
       |akka.remote.netty.tcp.port=$masterPort
      """.stripMargin
  )

  // 创建actor-SparkMaster
  val sparkMasterActorSystem = ActorSystem("SparkMaster", config)
  val sparkMaster: ActorRef = sparkMasterActorSystem.actorOf(Props[SparkMaster], "SparkMaster")

  // 给sparkMaster发送开始消息
  sparkMaster ! StartMessage(s"host:$masterHost port:$masterPort")
}
